package cn.com.cennavi.sparkStreamService;

import java.util.ArrayList;

import org.apache.spark.api.java.function.Function;
import org.apache.spark.api.java.function.PairFlatMapFunction;

import scala.Tuple2;

import com.cennavi.tymon.common.binparse.BinaryDataParseSTDAImpl;
import com.cennavi.tymon.common.binparse.bean.CNTF;
import com.cennavi.tymon.common.binparse.exception.BinPackageParseException;

import cn.com.cennavi.bean.IndexInputParmBean;

/**
 * 读取解析cntf
 * 
 * @Description
 * @author buf
 * @date 2017年3月9日 上午11:07:46
 */
public class IndexReadCNTFDataRDDStream implements PairFlatMapFunction<Tuple2<String, byte[]>, Long, IndexInputParmBean> {

	/** @Fields serialVersionUID: */

	private static final long serialVersionUID = 1L;
	private String CITYMESHID = null;

	public IndexReadCNTFDataRDDStream(String CITYMESHID) {
		this.CITYMESHID = CITYMESHID;
	}

	public Iterable<Tuple2<Long, IndexInputParmBean>> call(Tuple2<String, byte[]> arg0) throws Exception {
		// TODO Auto-generated method stub
		ArrayList<Tuple2<Long, IndexInputParmBean>> returnList = new ArrayList<Tuple2<Long, IndexInputParmBean>>();
		BinaryDataParseSTDAImpl cntfParse = new BinaryDataParseSTDAImpl();
		try {
			byte[] data = arg0._2;
			CNTF[] cntfs = cntfParse.parseLittleCNTF_G(data);
			for (CNTF cntf : cntfs) {
				// System.out.println(cntf.toKafkaString());
				if (this.CITYMESHID.contains(cntf.getAreaID() + ",")) {
					System.out.println(cntf.toKafkaString());
					IndexInputParmBean outdata = new IndexInputParmBean();
					long linkid = cntf.getFeatrueID();
					long time = cntf.getDataTime();
					int roadLength = cntf.getRoadLength();
					// 道路长度，值域[0,65535]。0表示未知，65535表示是无效值。
					if (roadLength == 0 || roadLength == 65535) {
						continue;
					}
					// cntf中旅行时间单位是0.1秒
					double travelTime = (double) cntf.getTravelTime() * 0.1;
					double speed = (double) roadLength / travelTime * 3.6;
					cntf.getLinkCarCnt();
					// computer speed:kmph
					outdata.linkId = linkid;
					outdata.speed = speed;
					outdata.timestamp = time;
					outdata.linkSimpleNum = 1;
					returnList.add(new Tuple2<Long, IndexInputParmBean>(linkid, outdata));
				}
			}
		} catch (BinPackageParseException e) {
			e.printStackTrace();
		}
		return returnList;
	}

}
